package en.edu.zxj.mq.mqserver.datacenter;

import en.edu.zxj.mq.common.MqException;
import en.edu.zxj.mq.mqserver.core.Binding;
import en.edu.zxj.mq.mqserver.core.Exchange;
import en.edu.zxj.mq.mqserver.core.MSGQueue;
import en.edu.zxj.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description：内存数据管理类 -- 实际消息转发/存储的类
 * 该类后续提供的一些方法, 可能会在多线程环境下使用, 因此需要注意线程安全的问题
 *
 * @author: zxj
 * @date: 2024-02-29
 * @time: 20:58:38
 */
@Slf4j
public class MemoryDataCenter {
    // key: 为 exchangeName, value: Exchange 对象
    private final ConcurrentHashMap<String, Exchange> exchangesMap = new ConcurrentHashMap<>();
    // key: 为 messageId, value: Message 对象
    private final ConcurrentHashMap<String, Message> messagesMap = new ConcurrentHashMap<>();
    // key: 为 exchangeName, value: Exchange 对象
    private final ConcurrentHashMap<String, MSGQueue> msgQueuesMap = new ConcurrentHashMap<>();
    // key1: exchangeName, key2: msgQueueName, value: Binding 对象
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    /**
     * 为了保证消息被正确消费了, 使用两种方式进行确认, 自动 ACK 和 手动 ACK
     * 其中自动 ACK 是指当信息被消费之后, 就会立即被销毁释放
     * 其中手动 ACK 是指当消息被消费之后, 有消费者主动调用一个 basicACK 方法,
     * 进行主动确认, 服务器收到这个确认后, 才能真正销毁消息
     * 此处的 "未确认消息" 就是指在手动 ACK 模式下, 该消息还没有被调用 basicACK,
     * 此时消息不能删除, 但是要和其他未消费的消息区分开, 于是另搞了个结构
     **/
    // key1: msgQueueName, value: Message 对象链表 -- 表示队列中有多少条消息(还未发送)
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // key1: msgQueueName, value: 依据 MessageId 存储 Message 对象 -- 表示队列中有多少条已经发送了, 但没有收到确认的消息, 使用 哈希表结构存储就是为了方便后续 ACK 后删除对应的消息
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAck = new ConcurrentHashMap<>();



    public void init() {
        // 目前无初始化逻辑
    }

    /**
     * 封装 Exchange 操作
     **/
    public void insertExchange(Exchange exchange) {
        exchangesMap.put(exchange.getName(), exchange);
        log.info("新交换机添加成功! exchangeName: {}", exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangesMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangesMap.remove(exchangeName);
        log.info("删除交换机成功! exchangeName: {}", exchangeName);
    }

    /**
     * 封装 MSGQueue 操作
     **/
    public void insertMSGQueue(MSGQueue msgQueue) {
        msgQueuesMap.put(msgQueue.getName(), msgQueue);
        log.info("新交换机添加成功! msgQueueName: {}", msgQueue.getName());
    }

    public MSGQueue getMSGQueue(String msgQueueName) {
        return msgQueuesMap.get(msgQueueName);
    }

    public void deleteMSGQueue(String msgQueueName) {
        msgQueuesMap.remove(msgQueueName);
        log.info("删除交换机成功! msgQueueName: {}", msgQueueName);
    }

    /**
     * 封装 Binding 操作
     **/
    public void insetBinding(Binding binding) throws MqException {
        // 传统的创建 Map 的步骤, 因为不是原子性操作, 存在线程安全的问题, 为了保证线程安全, 可以加上 synchronized 加锁
        // ConcurrentHashMap<String, Binding> stringBindingConcurrentHashMap = bindingsMap.get(binding.getExchangeName()) ;
        // if (stringBindingConcurrentHashMap == null) {
        //     stringBindingConcurrentHashMap = new ConcurrentHashMap<>();
        //     bindingsMap.put(binding.getExchangeName(),stringBindingConcurrentHashMap);
        // }
        // ConcurrentHashMap 中有提供了 computeIfAbsent 方法, 简化了上述步骤, 并且是线程安全的 --
        // 先使用 exchangeName 查询一下, 如果存在就直接返回, 如果不存在就创建
        ConcurrentHashMap<String, Binding> stringBindingConcurrentHashMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), (k) -> {
            return new ConcurrentHashMap<>();
        });


        synchronized (stringBindingConcurrentHashMap) {
            // 这里先查询在插入, 具有强的顺序关系, 数据存在二次覆盖,  存在线程安全的问题
            // 在根据 msgQueueName 查找, 如果存在, 就直接抛异常, 不存在才能插入
            if (stringBindingConcurrentHashMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在, exchangeName: " + binding.getExchangeName() +
                        "; msgQueueName: " + binding.getQueueName());
            }

            stringBindingConcurrentHashMap.put(binding.getQueueName(), binding);
        }

        log.info("绑定添加成功成功, binding.exchangeName: {}, binding.queueName: {},", binding.getExchangeName(), binding.getQueueName());
    }

    // 获取绑定
    // 1. 依据 exchangeName 和 queueName 获取唯一的 binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    // 2. 依据 exchangeName 获取所有的 binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列, 报错
            throw new MqException("[MemoryDataCenter] 绑定不存在! binding: " + binding);
        }

        bindingsMap.remove(binding.getExchangeName());
        log.info("删除绑定成功! binding: {}", binding);
    }


    /**
     * 封装信息操作
     **/
    // 添加信息
    public void addMessage(Message message) {
        messagesMap.put(message.getMessageId(), message);
        log.info("添加信息成功! messageId: {}", message.getMessageId());
    }

    // 依据 Id 查询信息
    public Message getMessage(String messageId) {
        return messagesMap.get(messageId);
    }

    // 依据 Id 删除信息
    public void deleteMessage(String messageId) {
        messagesMap.remove(messageId);
        log.info("消息被删除! messageId: {}", messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 把消息放到对应的队列数据结构中
        // 先根据队列的名字, 找到该队列对应的消息链表
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), (k) -> {
            return new LinkedList<>();
        });
        // 把数据假如到 messages 里面
        synchronized (messages) {
            messages.add(message);
        }
        // 在这里把该消息也往消息中心中插入一下, 假设如果 message 已经在消息中心存在, 重复插入也没有关系
        // 主要就是相同 messageId, 对应的 message 的内容一定是一样的. (服务器不会对 Message 内容修改 basicProperties 和 body)
        addMessage(message);
        log.info("消息被投递到队列中! messageId = " + message.getMessageId());
    }

    // 从队列中取消息
    public Message pollMessage(String queueName) {
        // 根据队列名, 查找一下, 对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 为空
        if (messages == null) {
            return null;
        }

        synchronized (messages) {
            // 队列中没有任何消息
            if (messages.isEmpty()) {
                return null;
            }
            // 链表中有元素, 就进行头删
            Message currentMessage = messages.remove(0);
            log.info("从消息从队列中取出! messageId: {}", currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        // 根据队列名, 查找一下, 对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }

        synchronized (messages) {
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAck.computeIfAbsent(queueName, (k) -> {
            return new ConcurrentHashMap<>();
        });
        messageHashMap.put(message.getMessageId(), message);
        log.info("消息进入待确认队列! messageId: {}", message.getMessageId());
    }

    // 删除未确认的消息(消息已经确认了)
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAck.computeIfAbsent(queueName, (k) -> {
            return new ConcurrentHashMap<>();
        });
        messageHashMap.remove(messageId);
        log.info("消息从待确认队列中删除! messageId: {}", messageId);
    }


    // 获取指定的未确认的信息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageConcurrentHashMap = queueMessageWaitAck.get(queueName);
        if (messageConcurrentHashMap == null) {
            return null;
        }
        return messageConcurrentHashMap.get(messageId);
    }


    // 这个方法就是从硬盘上读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中 -- 交换机, 消息队列, 绑定, 消息
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException, ClassNotFoundException {
        // 0. 清空之前的所有数据
        exchangesMap.clear();
        msgQueuesMap.clear();
        bindingsMap.clear();
        messagesMap.clear();
        queueMessageMap.clear();
        // 1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangesMap.put(exchange.getName(), exchange);
        }
        log.info("恢复所有的 交换机 数据成功!");
        // 2. 恢复所有的队列数据
        List<MSGQueue> msgQueues = diskDataCenter.selectAllMSGQueues();
        for (MSGQueue msgQueue : msgQueues) {
            msgQueuesMap.put(msgQueue.getName(), msgQueue);
        }
        log.info("恢复所有的 队列 数据成功!");
        // 3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), (k) -> {
                return new ConcurrentHashMap<>();
            });
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("恢复所有的 绑定 数据成功!");
        // 4. 恢复所有的消息队列
        //    遍历所有的队列, 根据每个队列的名字, 获取到所有的消息
        for (MSGQueue msgQueue : msgQueues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(msgQueue.getName());
            queueMessageMap.put(msgQueue.getName(), messages);
            for (Message message : messages) {
                messagesMap.put(message.getMessageId(), message);
            }
        }
        log.info("恢复所有的 消息队列 成功!");

        log.info("从磁盘中恢复所有数据到内存成功");

        // 规定:
        // 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复, 之前考虑硬盘存储的时候, 也没有设定这一块
        // 这个消息在硬盘上存储的时候, 就是当做 "未被取走"

    }



}

